package studio.raptor.databus.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.cmdb.ConfigFile;
import studio.raptor.cmdb.ConfigFileChangeListener;
import studio.raptor.cmdb.ConfigService;
import studio.raptor.cmdb.core.enums.ConfigFileFormat;
import studio.raptor.cmdb.model.ConfigFileChangeEvent;
import studio.raptor.databus.client.config.BrokerGroup;
import studio.raptor.databus.command.BizPubSubEventInfo;
import studio.raptor.databus.command.EventPubSubInfo;

import javax.jms.JMSException;
import java.io.*;
import java.util.*;
import studio.raptor.databus.command.EventTransportType;

/**
 * Created on 2017/5/25  <br>
 * <p>
 * 队列客户端配置类，包含连接服务端的基本信息，用户名，密码，服务URL<br>
 * <p>另持有常量：虚拟主题名称前缀、组消费目的地名称前缀</p>
 *
 * @author koalton
 */
public final class DatabusConfig {

  static final String LOAD_CONFIG_FAILURE = "load_failure";
  static final String VIRTUAL_TOPIC_PRE = "VirtualTopic.";
  static final String GROUP_QUEUE_NAME_PRE = "Consumer.";
  static final String CLIENT_CONFIG_PRE = "ClientCfg_";
  static final String EVENT_PUB_SUB_RELATION = "eventpubsub_relation";
  static final String BROKER_GROUP_CONFIG = "brokerconfig";
  static final String CONFIG_LOCAL_DIR = "databus-client/";

  /**
   * 协议传输默认选用文本格式传输
   */
  public static final EventTransportType TRANSPORTTYPE = EventTransportType.EVENT_TRANSPORT_TYPE_TEXT;


  private static Logger log = LoggerFactory.getLogger(DatabusConfig.class);
  /**
   * 死信队列名称前缀,默认为<code>DLQ.</code>
   */
  public static final String DLQ_PRE = "DLQ.";

  /**
   * 业务中心标志：消费者用此信息控制消费者组名，控制当前消费者是否支持所消费的事件序号；生产者,
   */
  private String bizKey;
  /**
   * 加载方式，配置中心或本地化
   */
  private DatabusConfigLoadType loadType = DatabusConfigLoadType.LOAD_TYPE_CMDB;

  /**
   * broker集合
   */
  private List<BrokerGroup> brokerGroups = new ArrayList<>(4);

  private BizPubSubEventInfo bizPubSubEventInfo = new BizPubSubEventInfo();

  /**
   * @param bizKey 业务中心
   */
  public DatabusConfig(String bizKey) throws JMSException, IOException {
    this.bizKey = bizKey;

    //load broker group info
    ConfigFile brokerGroupConfig =
        ConfigService.getConfigFile(DatabusConfig.BROKER_GROUP_CONFIG, ConfigFileFormat.JSON);
    if( null ==  brokerGroupConfig.getContent()){
      log.info("load config from CMDB failure ,turn to load from local .");
      //load from CMDB failure
      loadType = DatabusConfigLoadType.LOAD_TYPE_LOCAL;
      String content = loadLoaclBrokerGroupConfig();
      if( !"".equals(content) ){
        JSONArray jsonArray = JSON.parseArray(content);
        this.brokerGroups = jsonArray.toJavaList(BrokerGroup.class);
      }
    }else{
      log.info("load config from CMDB success ");
      String content = brokerGroupConfig.getContent();
      JSONArray jsonArray = JSON.parseArray(content);
      this.brokerGroups = jsonArray.toJavaList(BrokerGroup.class);
    }

    // load event pubsub info
    if (loadType == DatabusConfigLoadType.LOAD_TYPE_LOCAL) {
      String eventPubSubConfig = loadLoaclEventPubSubConfig();
      if (eventPubSubConfig != null && !"".equals(eventPubSubConfig)) {
        bizPubSubEventInfo = JSON.parseObject(eventPubSubConfig, BizPubSubEventInfo.class);
      }
    } else {
      //从配置中心加载事件发布、订阅关系配置--当且仅当使用配置中心时候，才实现热更新事件订阅关系
      bizPubSubEventInfo = loadCMDBEventPubSubConfig();
      //为配置中心增加监听
      listenCMDBEventPubSubConfig(bizPubSubEventInfo);
    }

  }


  public String getBizKey() {
    return bizKey;
  }

  public BizPubSubEventInfo getBizPubSubEventInfo() {
    return bizPubSubEventInfo;
  }

  DatabusConfigLoadType getLoadType() {
    return loadType;
  }

  /**
   * 加载本地化事件发布订阅配置
   *
   * @return 本地化事件订阅发布配置的字符串, 若无法找到配置文件返回空字符串
   */
  private String loadLoaclEventPubSubConfig() throws IOException {
    InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(CONFIG_LOCAL_DIR + "pubsubevent.json");
    BufferedReader bufferedReader = null;
    StringBuilder eventPubSub = new StringBuilder();
    if (in != null) {
      Reader reader = new InputStreamReader(in);
      try {
        bufferedReader = new BufferedReader(reader);
        String lineText = "";
        while ((lineText = bufferedReader.readLine()) != null) {
          eventPubSub.append(lineText);
        }
      } catch (IOException e) {
        e.printStackTrace();
      } finally {
        if (bufferedReader != null) {
          bufferedReader.close();
        }
        in.close();
      }
    }
    return eventPubSub.toString();
  }
  /**
   * 加载本地化Broker配置
   *
   * @return 本地化Broker配置, 若无法找到配置文件返回空字符串
   */
  private String loadLoaclBrokerGroupConfig() throws IOException {
    InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(CONFIG_LOCAL_DIR + "brokerconfig.json");
    BufferedReader bufferedReader = null;
    StringBuilder brokerGroupConfig = new StringBuilder();
    if (in != null) {
      Reader reader = new InputStreamReader(in);
      try {
        bufferedReader = new BufferedReader(reader);
        String lineText = "";
        while ((lineText = bufferedReader.readLine()) != null) {
          brokerGroupConfig.append(lineText);
        }
      } catch (IOException e) {
        e.printStackTrace();
      } finally {
        if (bufferedReader != null) {
          bufferedReader.close();
        }
        in.close();
      }
    }
    return brokerGroupConfig.toString();
  }


  /**
   * 从配置中心加载事件订阅发布配置
   *
   * @return 返回当前业务中心所发布、订阅的事件的配置
   */
  private BizPubSubEventInfo loadCMDBEventPubSubConfig() {
    BizPubSubEventInfo bizPubSubEventInfo = new BizPubSubEventInfo();
    ConfigFile eventPubSubConfigFile = ConfigService
        .getConfigFile(DatabusConfig.EVENT_PUB_SUB_RELATION, ConfigFileFormat.JSON);
    String eventPubSubConfigStr = eventPubSubConfigFile.getContent();
    HashMap<String, EventPubSubInfo> cfgs = JSON.parseObject(eventPubSubConfigStr, HashMap.class);
    Set<Map.Entry<String, EventPubSubInfo>> sets = cfgs.entrySet();
    Iterator<Map.Entry<String, EventPubSubInfo>> it = sets.iterator();
    while (it.hasNext()) {
      Map.Entry<String, EventPubSubInfo> entry = it.next();
      String eventNameSeq = entry.getKey();
      //如果eventName中本身包含了下划线
      int lastMark = eventNameSeq.lastIndexOf("_");
      if (lastMark == -1) {
        log.info("事件信息【" + eventNameSeq + "】未正确配置");
        continue;
      }
      String eventName = eventNameSeq.substring(0, lastMark);
      String eventSeq = eventNameSeq.substring(lastMark + 1, eventNameSeq.length());
      String objectStr = JSON.toJSONString(entry.getValue());
      EventPubSubInfo pubSubInfo = JSON.parseObject(objectStr, EventPubSubInfo.class);
      //当前事件被当前业务中心发布
      if (pubSubInfo.getPubBizCenterName().equals(bizKey)) {
        bizPubSubEventInfo.getPub().put(eventNameSeq, pubSubInfo.getBrokerGroupID());
        //this.pubEventBrokerMap.put(Integer.valueOf(eventSeq), pubSubInfo.getBrokerGroupID());
      }
      //当前事件被当前业务中心订阅
      if (pubSubInfo.getSubBizCenter().contains(bizKey)) {
        bizPubSubEventInfo.getSub().put(eventNameSeq,pubSubInfo.getBrokerGroupID());
        //this.subEventBrokerMap.put(Integer.valueOf(eventSeq), pubSubInfo.getBrokerGroupID());
      }
    }
    return bizPubSubEventInfo;
  }

  /**
   * 监听配置中心事件订阅、发布配置变化
   *
   * @param bizPubSubEventInfo 业务中心的事件订阅发布关系，当监听到配置中心发生改变时，会对此值进行修改
   */
  private void listenCMDBEventPubSubConfig(final BizPubSubEventInfo bizPubSubEventInfo) {
    final String tempBizKey = bizKey;
    ConfigFile eventPubSubConfigFile = ConfigService
        .getConfigFile(DatabusConfig.EVENT_PUB_SUB_RELATION, ConfigFileFormat.JSON);
    eventPubSubConfigFile.addChangeListener(new ConfigFileChangeListener() {
      @Override
      public void onChange(ConfigFileChangeEvent configFileChangeEvent) {
        log.info("The event pub-sub relation has been changed,old value is " + configFileChangeEvent
            .getOldValue() + ",new value is " + configFileChangeEvent.getNewValue());
        String newContent = configFileChangeEvent.getNewValue();
        HashMap<String, EventPubSubInfo> cfgs = JSON.parseObject(newContent, HashMap.class);
        Set<Map.Entry<String, EventPubSubInfo>> sets = cfgs.entrySet();
        Iterator<Map.Entry<String, EventPubSubInfo>> it = sets.iterator();
        synchronized (bizPubSubEventInfo) {
          while (it.hasNext()) {
            Map.Entry<String, EventPubSubInfo> entry = it.next();
            String eventNameSeq = entry.getKey();
            String eventName = eventNameSeq.split("[_]")[0];
            String eventSeq = eventNameSeq.split("[_]")[1];
            String objectStr = JSON.toJSONString(entry.getValue());
            EventPubSubInfo pubSubRelation = JSON.parseObject(objectStr, EventPubSubInfo.class);
            //当前事件被当前业务中心发布
            if (pubSubRelation.getPubBizCenterName().equals(tempBizKey)) {
              bizPubSubEventInfo.getPub().put(eventNameSeq, pubSubRelation.getBrokerGroupID());
            }
            //当前事件被当前业务中心订阅
            if (pubSubRelation.getSubBizCenter().contains(tempBizKey)) {
              bizPubSubEventInfo.getSub().put(eventNameSeq, pubSubRelation.getBrokerGroupID());
            }

          }
        }
      }
    });

  }

  List<BrokerGroup> getBrokerGroups() {
    return brokerGroups;
  }

  void setBrokerGroups(List<BrokerGroup> brokerGroups) {
    this.brokerGroups = brokerGroups;
  }

}

